package club.hicode.daydayup.zookeeper;

import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.RandomUtil;
import com.github.sd4324530.jtuple.Tuple2;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.api.CuratorListener;
import org.apache.curator.framework.api.UnhandledErrorListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.framework.recipes.locks.InterProcessSemaphoreV2;
import org.apache.curator.framework.recipes.locks.Lease;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.retry.RetryForever;
import org.apache.curator.retry.RetryNTimes;
import org.apache.curator.retry.RetryUntilElapsed;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.data.Stat;
import org.junit.Test;

import java.util.List;
import java.util.Objects;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * Zookeeper 的测试
 *
 * @author Liu Chunfu
 * @date 2018-06-01 上午9:31
 **/
@Slf4j
public class TestZK {

    private static final String CONNECT_STR = "39.108.231.92:2181";
    //private static final String CONNECT_STR = "127.0.0.1:2181";

    @Test
    public void curator1() throws Exception {
        //配置连接
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString("127.0.0.1:2181")
                .sessionTimeoutMs(5000)
                .retryPolicy(retryPolicy)
                .build();

        String rootPath = "/zkStudy";
        String path = "/zkStudy/studyChildNode";
        String val = "this.is a zk node val";

        //启动
        client.start();

        //监听
        PathChildrenCache cache = new PathChildrenCache(client, rootPath, true);
        cache.start();
        //设置监听器
        PathChildrenCacheListener listener = (listenerClient, event) -> {
            System.out.println("event happen : " + event.getType());
            if (null != event.getData()) {
                System.out.println("node path : " + event.getData().getPath() + "   node val : " + new String(event.getData().getData()));
            }
        };
        //将监听器加入监听队列
        cache.getListenable().addListener(listener);

        //创建
        client.create()
                .creatingParentContainersIfNeeded()
                .withMode(CreateMode.PERSISTENT)
                .forPath(path, val.getBytes());

        //读取
        Stat stat = new Stat();
        byte[] dataBytes = client.getData().storingStatIn(stat).forPath(path);
        String readVal = new String(dataBytes);
        System.out.println(readVal);

        //更新
        String newVal = "this ia a new zk node val";
        client.setData().forPath(path, newVal.getBytes());

        //判断是否存在
        Stat existFlagStat = client.checkExists().forPath("/notExistPath");
        System.out.println("node /notExistPath is exist : " + !Objects.isNull(existFlagStat));

        //删除
        client.delete()
                .deletingChildrenIfNeeded()
                .forPath(rootPath);
    }


    @Test
    public void curator2() throws Exception {
        //必须设置重连策略，否则会出现问题
        CuratorFramework framework = CuratorFrameworkFactory.builder().connectString(CONNECT_STR)
                .retryPolicy(new RetryNTimes(14, 1500)).build();

        //必须先启动
        framework.start();

        String s = framework.create().creatingParentContainersIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/mic/core", "127.0.0.1".getBytes());

        System.out.println(s);
    }

    @Test
    public void curator3() throws Exception {

        CuratorFramework client = CuratorFrameworkFactory.builder().retryPolicy(new RetryForever(5000)).connectString("127.0.0.1:2181")
                .build();
        client.start();
        List<String> deep1 = client.getChildren().forPath("/");
        List<Tuple2<String, List<String>>> collect = deep1.stream().map(d1 -> {
            try {
                System.out.println(d1);
                return Tuple2.with(d1, client.getChildren().forPath("/" + d1));
            } catch (Exception e) {
                e.printStackTrace();
            }
            return null;
        }).collect(Collectors.toList());

        collect.forEach(t2 -> System.out.println(t2.first + ":" + t2.second));

        client.close();
    }

    @Test
    public void curator4() throws Exception {
        CuratorFramework client = CuratorFrameworkFactory.builder().retryPolicy(new RetryNTimes(4, 5000)).connectString(CONNECT_STR).build();
        client.start();
        //父节点也删除
        client.delete().deletingChildrenIfNeeded().forPath("/mic");

        System.out.println("success");
    }

    /**
     * 创建
     *
     * @throws Exception
     */
    @Test
    public void curatorCreate() throws Exception {
        CuratorFramework client = CuratorFrameworkFactory.builder().retryPolicy(new RetryUntilElapsed(100, 20))
                .connectString(CONNECT_STR).build();
        client.start();
        String s = client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath("/mic/hello", "129.123.1.1:8933".getBytes());
        System.out.println(s);
    }

    /**
     * 创建
     *
     * @throws Exception
     */
    @Test
    public void curatorGet() throws Exception {
        String nodePath = "/mic-core/consumer";

        CuratorFramework client = CuratorFrameworkFactory.builder().retryPolicy(new RetryUntilElapsed(100, 20)).connectString(CONNECT_STR).build();
        client.start();

        client.setData().forPath(nodePath, "Hello world!".getBytes());
        System.out.println("set success");

        Stat stat = client.checkExists().forPath(nodePath);
        if (stat == null) {
            System.out.println("NULL");
            return;
        }

        client.setData().forPath(nodePath, "Hello world!222".getBytes());
        byte[] bytes = client.getData().forPath(nodePath);
        String str = new String(bytes);

        System.out.println(str);
    }

    @Test
    public void listenerTest() throws Exception {
        CuratorFramework client = CuratorFrameworkFactory.builder().retryPolicy(new RetryUntilElapsed(100, 20)).connectString(CONNECT_STR).build();
        client.start();

        ConnectionStateListener connectionStateListener = new ConnectionStateListener() {
            @Override
            public void stateChanged(CuratorFramework client, ConnectionState newState) {
                System.err.println("状态变更。。，当前状态：" + newState);
            }
        };

        client.getConnectionStateListenable().addListener(connectionStateListener);
        client.getUnhandledErrorListenable().addListener(new UnhandledErrorListener() {
            @Override
            public void unhandledError(String message, Throwable e) {
                System.err.println(message + ":" + e.getMessage());
            }
        });

        //client.getCuratorListenable().addListener(new CuratorListener() {
        //    @Override
        //    public void eventReceived(CuratorFramework client, CuratorEvent event) throws Exception {
        //        event.
        //    }
        //});
    }

    @Test
    public void curatorListener1() throws Exception {
        CuratorFramework client = CuratorFrameworkFactory.builder().retryPolicy(new RetryNTimes(5, 2000))
                .connectString(CONNECT_STR).build();
        client.start();

        String path = "/hicode/club";


        ////创建节点 需要注意，判定该节点是否存在，否则会报错：org.apache.zookeeper.KeeperException$NodeExistsException: KeeperErrorCode = NodeExists for /hicode/club
        //client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(path, "Love".getBytes());
        //ThreadUtil.sleep(2, TimeUnit.SECONDS);

        //需要注意，此监听的必须是已经存在的服务节点，否则会报错： org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /hicode/club2
        client.getData().usingWatcher(new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                System.out.println("监听器watchedEvent：" + watchedEvent);
            }
        }).forPath(path);

        //设置获取修改节点
        System.out.println("第一次改变");
        client.setData().forPath(path, "Hicode".getBytes());
        client.getData().usingWatcher(new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                System.out.println("监听器watchedEvent：" + watchedEvent);
            }
        }).forPath(path);
        ThreadUtil.sleep(2, TimeUnit.SECONDS);


        System.out.println("第二次改变");
        client.setData().forPath(path, "Club".getBytes());
        client.getData().usingWatcher(new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                System.out.println("监听器watchedEvent：" + watchedEvent);
            }
        }).forPath(path);
        ThreadUtil.sleep(2, TimeUnit.SECONDS);

        System.out.println("第三次改变");
        client.setData().forPath(path, "Club3".getBytes());
        ThreadUtil.sleep(100, TimeUnit.SECONDS);
    }


    @Test
    public void curatorListener2() throws Exception {
        CuratorFramework client = CuratorFrameworkFactory.builder().retryPolicy(new RetryNTimes(5, 2000))
                .connectString(CONNECT_STR).build();
        client.start();
        String path = "/hicode/club";

        CuratorListener listener = new CuratorListener() {
            @Override
            public void eventReceived(CuratorFramework client, CuratorEvent event) throws Exception {
                System.out.println("监听事件触发，event事件为：" + event.getType());
            }
        };

        client.getCuratorListenable().addListener(listener);

        //创建临时子节点
        client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).inBackground().forPath(path + "/hi");
// 异步获取节点数据
        client.getData().inBackground().forPath(path);
// 变更节点内容inBackground 。
        client.setData().forPath(path, "124443".getBytes());
//创建临时子节点
        client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(path + "/hi");
        client.delete().deletingChildrenIfNeeded().forPath(path);

        //
        ////创建临时子节点
        //client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).inBackground().forPath(path + "/hi");
        //// 异步获取节点数据
        //client.getData().inBackground().forPath(path);
        //// 变更节点内容inBackground 。
        //client.setData().inBackground().forPath(path, "123".getBytes());
        //// 变更节点内容inBackground 。
        //client.setData().forPath(path, "124443".getBytes());
        ////创建临时子节点
        //client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).inBackground().forPath(path + "/hi");
        //client.delete().deletingChildrenIfNeeded().inBackground().forPath(path);

        ThreadUtil.sleep(1000, TimeUnit.SECONDS);
    }

    @Test
    public void showChild() throws Exception {
        CuratorFramework client = CuratorFrameworkFactory.builder().connectString(CONNECT_STR).retryPolicy(new RetryNTimes(5, 100000)).build();
        client.start();


        List<String> strings = client.getChildren().forPath("/");
        strings.forEach(System.out::println);

    }

    @Test
    public void createLockPath() throws Exception {
        CuratorFramework client = CuratorFrameworkFactory.builder().connectString(CONNECT_STR).retryPolicy(new RetryNTimes(5, 100000)).build();
        client.start();

        String s = client.create().creatingParentsIfNeeded().forPath("/lock");
        System.out.println(s);
    }

    @Test
    public void testLock() throws Exception {
        CuratorFramework client = getClient();
        InterProcessMutex lock = new InterProcessMutex(client, "/lock");
        lock.acquire();

        System.out.println("lock...");
        lock.release();
    }

    @Test
    public void testLock1() throws Exception {
        CuratorFramework client = getClient();

        ExecutorService executorService = Executors.newFixedThreadPool(5);
        for (int i = 0; i < 10; i++) {
            executorService.execute(() -> {
                try {
                    doWork(client);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        }
        ThreadUtil.sleep(10, TimeUnit.MINUTES);
    }


    private void doWork(CuratorFramework client) throws Exception {
        InterProcessMutex lock = new InterProcessMutex(client, "/lock");
        try {
            lock.acquire();
            System.out.println("current thread is " + Thread.currentThread().getId() + " 获取锁...");
            ThreadUtil.sleep(RandomUtil.randomInt(5), TimeUnit.SECONDS);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            System.out.println("current thread is " + Thread.currentThread().getId() + " 释放锁...");
            lock.release();
        }
    }


    @Test
    public void testxx() {
        ExecutorService executorService = Executors.newFixedThreadPool(3);
        for (int i = 0; i < 10; i++) {
            executorService.execute(() -> {
                try {
                    doWorkSemaphore(getClient());
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        }
        ThreadUtil.sleep(10, TimeUnit.MINUTES);
    }


    private void doWorkSemaphore(CuratorFramework client) throws Exception {
        InterProcessSemaphoreV2 lock = new InterProcessSemaphoreV2(getClient(), "/path", 2);

        Lease acquire = lock.acquire();
        try {
            System.out.println("current thread is " + Thread.currentThread().getId() + " 获取锁...");
            ThreadUtil.sleep(RandomUtil.randomInt(5), TimeUnit.SECONDS);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            System.out.println("current thread is " + Thread.currentThread().getId() + " 释放锁...");
            lock.returnLease(acquire);
        }
    }

    private CuratorFramework getClient() {
        CuratorFramework client = CuratorFrameworkFactory.builder().connectString(CONNECT_STR).retryPolicy(new RetryNTimes(5, 100000)).build();
        client.start();
        return client;
    }
}
